package com.zengjianjun.canal.demo.ordinary;

import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 多线程消费
 *
 * @author zengjianjun
 */
@Slf4j
@Component
public class CanalUserConsumer implements CommandLineRunner {

    @Value("${canal.hostname:192.168.137.128}")
    private String canalHostname;
    @Value("${canal.port:11111}")
    private int canalPort;
    @Value("${canal.subscribe}")
    private String canalSubscribe;
    @Value("${canal.destination}")
    private String canalDestination;

    // 创建一个 PriorityBlockingQueue 用于按照 batchId 排序批次数据
    private static final PriorityBlockingQueue<BatchIdInfo> BATCH_ID_QUEUE = new PriorityBlockingQueue<>(100, Comparator.comparing(BatchIdInfo::getBatchId));

    private volatile CanalConnector connector;

    private static final int THREAD_NUM = 4;

    @Override
    public void run(String... args) throws Exception {
//        new Thread(this::start).start();
    }

    //    @PostConstruct
    public void init() {
        if (connector == null) {
            synchronized (CanalUserConsumer.class) {
                if (connector == null) {
                    log.info("canalHostname = {}, canalPort = {}, canal.subscribe = {}, canal.destination = {}", canalHostname, canalPort, canalSubscribe, canalDestination);
                    // 创建连接，配置连接参数
                    connector = CanalConnectors.newSingleConnector(
                            new InetSocketAddress(canalHostname, canalPort),
                            canalDestination,
                            "",
                            ""
                    );
                    connector.connect();
                    // 设置订阅规则
                    connector.subscribe(canalSubscribe);
                    // 从头开始消费
                    connector.rollback();
                }
            }
        }
        // 消息确认
        this.ackBatchId();
    }

    public void start() {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUM);
        for (int i = 0; i < THREAD_NUM; i++) {
            executorService.execute(() -> {
                while (true) {
                    try {
                        Message message = connector.getWithoutAck(1000, 1L, TimeUnit.SECONDS);
                        long batchId = message.getId();
                        if (batchId != -1 && !message.getEntries().isEmpty()) {
                            log.info("batchId = {}, size = {}", batchId, message.getEntries().size());
                            messageProcess(message);
                            BATCH_ID_QUEUE.add(BatchIdInfo.builder().threadId(Thread.currentThread().getName()).batchId(batchId).build());
                        }
                        Thread.sleep(50);
                    } catch (Exception e) {
                        log.error("canal get message error: ", e);
                    }
                }
            });
        }
    }

    /**
     * canal消息消费批次确认
     */
    public void ackBatchId() {
        new Thread(() -> {
            log.info("canal ack batchId start...");
            while (true) {
                try {
                    // 保证每个线程都有add数据进入BATCH_ID_QUEUE中，否则ack确认时可能顺序错乱，导致canal多线程消费异常
                    int threadGroupSize = Arrays.stream(BATCH_ID_QUEUE.toArray(new BatchIdInfo[0])).collect(Collectors.groupingBy(BatchIdInfo::getThreadId)).size();
                    if (threadGroupSize < THREAD_NUM) {
                        Thread.sleep(300L);
                        continue;
                    }
                    // poll获取元素，take阻塞获取元素
                    Long batchId = BATCH_ID_QUEUE.take().getBatchId();
                    log.info("canal ack batchId = {}", batchId);
                    connector.ack(batchId);
                } catch (Exception e) {
                    log.error("canal ack batchId error: ", e);
                }
            }
        }).start();
    }

    /**
     * 消息处理
     *
     * @param message 消息数据
     */
    private void messageProcess(Message message) {
        try {
            // 数据过滤
            List<CanalEntry.Entry> list = message.getEntries().stream().filter(data -> {
                try {
                    CanalEntry.EntryType entryType = data.getEntryType();
                    if (entryType == CanalEntry.EntryType.TRANSACTIONBEGIN || entryType == CanalEntry.EntryType.TRANSACTIONEND) {
                        // 过滤开启/关闭事务操作类型
                        return false;
                    }
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(data.getStoreValue());
                    // 获取操作类型：insert/update/delete类型
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    if (eventType != CanalEntry.EventType.INSERT) {
                        // 非新增类型
                        return false;
                    }
                } catch (Exception e) {
                    log.error("message filter error: data = {}", JSONUtil.toJsonStr(data), e);
                    return false;
                }
                return true;
            }).collect(Collectors.toList());
            // 数据解析
            int size = list.size();
            int batchSize = 0;
            for (int i = 0; i < size; i++) {
                CanalEntry.Entry entry = list.get(i);
                // 打印当前批次最后一条Header信息
                if (i == size - 1) {
                    log.info(String.format("binlog[%s:%s] , dbName:%s, tableName:%s ",
                            entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getSchemaName(), entry.getHeader().getTableName()));
                }
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
                batchSize += rowDataList.size();
                for (int j = 0; j < rowDataList.size(); j++) {
                    CanalEntry.RowData rowData = rowDataList.get(j);
                    List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
                    Map<String, CanalEntry.Column> columnMap = columnList.stream().collect(Collectors.toMap(CanalEntry.Column::getName, Function.identity()));
                    CanalEntry.Column idColum = columnMap.get("id");
                    CanalEntry.Column usernameColumn = columnMap.get("username");
                    CanalEntry.Column passwordColumn = columnMap.get("password");
                    CanalEntry.Column createTimeColumn = columnMap.get("create_time");
                    // 打印当前批次总共处理的数据行数及最后一行数据的信息
                    if (j == rowDataList.size() - 1 && i == size - 1) {
                        log.info("batchSize = {}, id = {}, createTime = {}", batchSize, this.getColumnValue(idColum), this.getColumnValue(createTimeColumn));
                    }
                }
            }
        } catch (Exception e) {
            log.info("canal message process error: ", e);
        }
    }

    public String getColumnValue(CanalEntry.Column column) {
        return column == null ? null : column.getValue();
    }

    @Data
    @Builder
    static class BatchIdInfo {
        /**
         * 线程ID
         */
        private String threadId;
        /**
         * 批次ID
         */
        private Long batchId;
    }
}
